package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.process.IHotArticleProcessor;
import com.heima.behavior.dto.ArticleAggregatorResult;
import com.heima.behavior.dto.ArticleOperationMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

import java.time.Duration;

/**
 * @author: itheima
 * @create: 创建XXXProcesser接口接口里声明@Input("input")源头话题 @Output("output")目标话题 --2 @EnableBinding(XXXProcesser.class)编写流式计算监听器 @StreamListener("input")  接收到业务编写核心业务将计算结果	发送到 @SendTo("output") 目标话题   这些步骤从articel模块的process包开始了
 */
@Slf4j
@Component
@EnableBinding(IHotArticleProcessor.class) //此接口在process包下  绑定的是声明源头和目标话题的接口
public class HotArticleListener {

    /**
     * 实时统计所有文章操作行为得出文章对应 点赞数，阅读数，评论数，收藏数量
     *
     * @param input 数据流 形式：Key:null  val：{"add":1,"articleId":1553958754792316930,"type":1}
     * @return 统计分析结果 形式：{"articleId":1553958754792316930,"view":100,"like":50,"comment":10,"collect":10} 聚合的结果类型   统计次数 从上一行数据流中聚合得到的
     */
    @StreamListener("hot_article_score_topic") //和IHotArticleProcessor接口里定义的一致
    @SendTo("hot_article_result_topic") //和IHotArticleProcessor接口里定义的一致
    public KStream<String, String> process(KStream<String, String> input) {
        //1.处理原始数据 对Key进行重新指定 key不是null了 value是原样  处理后：Key：文章Id  Val：{"add":1,"articleId":1553958754792316930,"type":1} 是JSON字符串
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             *
             * @param key 数据流中键  null
             * @param value 数据流中值  {"add":1,"articleId":1553958754792316930,"type":1}
             * @return 新键值对象
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                log.info("1.实时监听到操作行为数据：{}", value);
                //1.1 对文章操作行为消息转为Java对象 ArticleOperationMessage  因为要想取{"add":1,"articleId":1553958754792316930,"type":1}这里数据的值，又因为此数据是（行为模块的实现类ApLikesBehaviorServiceImpl里的）ArticleOperationMessage对象转成的数据，so反序列化 行为模块的实现类ApLikesBehaviorServiceImpl里的）ArticleOperationMessage再转为java对象才能取
                ArticleOperationMessage articleOperationMessage = JSON.parseObject(value, ArticleOperationMessage.class); //将前面的转成后面的对象 value转成ArticleOperationMessage对象
                //1.2 重新设置数据键值
                log.info("1.得到新键值对象中Key：{}， Val：{}", articleOperationMessage.getArticleId(), value);
                return new KeyValue<>(articleOperationMessage.getArticleId() + "", value); //因为要求key和value是字符串string  so+“ ” 或.toString()也可    key从Null变成了文章id value保持不变
            }    //返回给37行第三个参数
        });
        //2.对数据里中记录根据Key进行分组     将文章id相同的数据相同的记录放在一组
        KGroupedStream<String, String> groupedStream = map.groupByKey();

        //3.设置时间窗口 统计：最近30S内操作行为
        TimeWindowedKStream<String, String> timeWindowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(30)));

        //4.设置聚合,将文章ID相同的所有文章操作行为全部获取到 前提是在设置的时间窗口之内            val是JSON字符串不能向之前那样设置完时间窗口后直接count计算，因为val是JSON字符串 Val：{"add":1,"articleId":1553958754792316930,"type":1}无法累加so需自己写此逻辑so 要写聚合
        //4.1 创建初始化聚合结果
        Initializer<String> initializer = new Initializer() { //是匿名内部类so alt enter 点提示的 implement method实现方法 就出现下面的@Override 的方法了
            @Override
            public String apply() {
                //初次执行聚合，第一次聚合结果
                log.info("4.初次聚合设置为null");
                return null;
            }
        };
        //4.2 创建聚合对象-完成聚合逻辑
        Aggregator<String, String, String> aggregator = new Aggregator<String, String, String>() {//是匿名内部类so alt enter 点提示的 implement method实现方法 就出现下面的@Override 的方法了
            /**
             * 前提：一个时间窗口内，将上次聚合结果跟本次聚合结果合并在一起
             * @param key 数据流Key  文章ID
             * @param value 数据流中Val 操作行为JSON {"add":1,"articleId":1553958754792316930,"type":1}
             * @param aggregate 上次聚合结果 {"articleId":1553958754792316930,"view":100,"like":50,"comment":10,"collect":10}
             * @return 将本次操作行为跟上次聚合结果累加
             */
            @Override
            public String apply(String key, String value, String aggregate) {
                log.info("4.2.0.进行聚合，上次聚合结果：{}", aggregate);
                //4.2.1 先获取上次聚合结果aggregate 有可能上次聚合为空     聚合累加是add 数字 和聚合的数字相加减
                ArticleAggregatorResult result = null;
                if (StringUtils.isBlank(aggregate)) {
                    //上次聚合结果为空 则设默认值0
                    result = new ArticleAggregatorResult();
                    result.setArticleId(Long.valueOf(key));
                    result.setView(0);//阅读  设默认值是0
                    result.setLike(0);//点赞
                    result.setCollect(0);//收藏
                    result.setComment(0);//评论
                } else {
                    //说明上次聚合结果有值 就得赋值                下面大概思路是 上次聚合结果有值 得跟本次对文章操作行为的增量值累加
                    result = JSON.parseObject(aggregate, ArticleAggregatorResult.class);
                }
                //4.2.2 在获取本次对文章操作行为
                ArticleOperationMessage articleOperationMessage = JSON.parseObject(value, ArticleOperationMessage.class);

                //4.2.3 将本次操作行为跟上次聚合结果累加
                Integer type = articleOperationMessage.getType();
                Integer add = articleOperationMessage.getAdd(); //本次的增量值
                switch (type) {
                    case 0:
                        result.setView(result.getView() + add);//上次聚合的查阅 +本次的增量值
                        break;
                    case 1:
                        result.setLike(result.getLike() + add);//上次聚合的点赞 +本次的增量值
                        break;
                    case 2:
                        result.setComment(result.getComment() + add);//上次聚合的评论 +本次的增量值
                        break;
                    case 3:
                        result.setCollect(result.getCollect() + add);//上次聚合的收藏 +本次的增量值
                        break;
                }
                log.info("4.2.3.进行聚合后结果：{}", JSON.toJSONString(result));
                return JSON.toJSONString(result);
            }
        };

        KTable<Windowed<String>, String> aggregate = timeWindowedKStream.aggregate(initializer, aggregator);

        //5.重新设置数据流Key跟Val                                       key 类型是Windowed<String> windowed类型  value是string     新键值对象类型  是string
        KStream<String, String> stream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {//是匿名内部类so alt enter 点提示的 implement method实现方法 就出现下面的@Override 的方法了
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
        return stream;
    }
}
//拿到聚合结果后将数据发送到目标话题中 重新设置key value       最终key依旧是articleId 但value变统计结果了